package com.dodayum.kafka.learnkafka.controller;

import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.producer.RecordMetadata;
import com.google.common.base.Throwables;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.StringUtils;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.servlet.http.HttpServletRequest;
import java.util.Map;
import java.util.concurrent.ExecutionException;

@RestController
public class KafkaProducerController {

    @Autowired
    KafkaTemplate<Integer, String> template;

    /**
     * 同步发消息
     * @return true
     */
    @RequestMapping("sendMessage/sync")
    public String send(HttpServletRequest request) {
        Map<String, String[]> parameterMap = request.getParameterMap();
        String requestData = JSON.toJSONString(parameterMap);
        if (StringUtils.isEmpty(requestData)){
            requestData = "没有发送任何数据";
        }
        ListenableFuture<SendResult<Integer, String>> future
                = template.send("topic-spring-01", 0, requestData);
        try {
            SendResult<Integer, String> sendResult = future.get();
            RecordMetadata recordMetadata = sendResult.getRecordMetadata();
            System.out.println("消息发送到了主题：" + recordMetadata.topic()
                    + "\n分区：" + recordMetadata.partition()
                    + "\n偏移量：" + recordMetadata.offset());
        } catch (InterruptedException e) {
            e.printStackTrace();
            return "报错了" + Throwables.getStackTraceAsString(e);
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        return "success";
    }

    /**
     * 异步发消息
     * @return true
     */
    @RequestMapping("sendMessage/Async")
    public String sendAsync(){
        ListenableFuture<SendResult<Integer, String>> future
                = template.send("topic-spring-01", 0, 1, "hello kafka by async");
        future.addCallback(
                new ListenableFutureCallback<SendResult<Integer, String>>() {
                    @Override
                    public void onFailure(Throwable ex) {
                        System.out.println("发送消息失败" + Throwables.getStackTraceAsString(ex));
                    }

                    @Override
                    public void onSuccess(SendResult<Integer, String> result) {
                        RecordMetadata recordMetadata = result.getRecordMetadata(); // 发送的愿数据
                        System.out.println("发送成功");
                        System.out.println("消息发送到了主题：" + recordMetadata.topic()
                                + "\n分区：" + recordMetadata.partition()
                                + "\n偏移量：" + recordMetadata.offset());
                    }
                }
        );
        return "true";
    }


}
